package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPreprocessApp extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeOrderPreprocessApp().init(2006,"DwdTradeOrderPreprocessApp");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        readOdsDb(tEnv,"DwdTradeOrderPreprocessApp");

        //过滤出订单明细表order_detail
        Table orderDetail = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['course_id'] course_id, " +
                " `data`['course_name'] course_name, " +
                " `data`['order_id'] order_id, " +
                " `data`['user_id'] user_id, " +
                " `data`['origin_amount'] origin_amount, " +
                " `data`['coupon_reduce'] coupon_reduce, " +
                " `data`['final_amount'] final_amount, " +
                " `data`['session_id'] session_id, " +
                " `data`['create_time'] create_time, " +
                " `data`['update_time'] update_time, " +
                " ts, " +
                " xid " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert'");
        //orderDetail.printSchema();
        tEnv.createTemporaryView("order_detail",orderDetail);

        //过滤出订单信息表order_info
        Table orderInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['origin_amount'] origin_amount, " +
                " `data`['coupon_reduce'] coupon_reduce, " +
                " `data`['final_amount'] final_amount, " +
                " `data`['order_status'] order_status, " +
                " `data`['out_trade_no'] out_trade_no, " +
                " `data`['trade_body'] trade_body, " +
                " `data`['session_id'] session_id, " +
                " `data`['province_id'] province_id, " +
                " `data`['create_time'] create_time, " +
                " `data`['expire_time'] expire_time, " +
                " `data`['update_time'] update_time, " +
                "`old`, " +
                "`type`, " +
                " ts, " +
                " xid " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info'");
        //orderInfo.printSchema();
        tEnv.createTemporaryView("order_info",orderInfo);

        //两表Join
        Table resultTable = tEnv.sqlQuery("select " +
                " od.id, " +
                " od.course_id, " +
                " od.course_name, " +
                " od.order_id, " +
                " od.user_id, " +
                " od.origin_amount, " +
                " od.coupon_reduce, " +
                " od.final_amount, " +
                " od.session_id, " +
                " od.create_time, " +
                " od.update_time, " +
                " oi.order_status, " +
                " oi.out_trade_no, " +
                " oi.trade_body, " +
                " oi.province_id, " +
                " oi.expire_time, " +
                " od.ts od_ts, " +
                " od.xid, " +
                " oi.`old`, " +
                " oi.`type`, " +
                " oi.ts oi_ts " +
                "from order_detail od join order_info oi on od.order_id=oi.id");
        //resultTable.printSchema();

        tEnv.executeSql("create table dwd_trade_order_app(" +
                " id string, " +
                " course_id string, " +
                " course_name string, " +
                " order_id string, " +
                " user_id string, " +
                " origin_amount string, " +
                " coupon_reduce string, " +
                " final_amount string, " +
                " session_id string, " +
                " create_time string, " +
                " update_time string, " +
                " order_status string, " +
                " out_trade_no string, " +
                " trade_body string, " +
                " province_id string, " +
                " expire_time string, " +
                " od_ts string, " +
                " xid string, " +
                " `old` map<string,string>, " +
                " `type` string, " +
                " oi_ts string" +
                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TRADE_ORDER_APP));
        resultTable.executeInsert("dwd_trade_order_app");


    }
}